/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity;

import java.math.BigDecimal;
import java.math.MathContext;
import java.util.Random;

import org.junit.Test;
import org.sabayframework.activity.impl.Fork;
import org.sabayframework.activity.impl.Pipeline;
import org.sabayframework.activity.impl.ReplicableActivity;
import org.sabayframework.activity.impl.StatefulActivity;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.channel.ram.ArrayChannel;
import org.sabayframework.channel.ram.ConcurrentArrayChannel;
import org.sabayframework.queue.ExecutionQueue;
import org.sabayframework.queue.impl.CentralizedExecutionQueue;


/**
 * Estimation of Pi using Monte Carlo method.
 * http://en.wikipedia.org/wiki/Monte_Carlo_method
 * @author Alex Antonau
 * */
public class MonteCarloPi {

	static final class WorkLoad{
		final Random random;
		final long size; //number of points to test
		long result; // points in circle
		public WorkLoad(Random random, long size) {
			super();
			this.random = random;
			this.size = size;
		}
		
	}

	public static final class PointCounter extends ReplicableActivity<WorkLoad, WorkLoad>
	{
		private WorkLoad workLoad;
		@Override
		public void input(ActivityContext<WorkLoad, WorkLoad> ctx, Lock<WorkLoad> inputLock) {
			workLoad = inputLock.get(0);
		}
		
		@Override
		public void process(ActivityContext<WorkLoad, WorkLoad> ctx) throws Exception {
			final Random random = workLoad.random;
			long count = 0;
			final long size = workLoad.size;
			for(long i=0;i<size;++i){
				final double x = random.nextDouble();
				final double y = random.nextDouble();
				if (x*x + y*y < 1.0) ++ count;
			}
			workLoad.result = count;
		}

		@Override
		public void output(ActivityContext<WorkLoad, WorkLoad> ctx,	Lock<WorkLoad> outputLock) {
			outputLock.set(0, workLoad);
		}
		@Override
		public ReplicableActivity<WorkLoad, WorkLoad> replicate() {
			return new PointCounter();
		}
	}
	
	public static class PiCalculator extends StatefulActivity<WorkLoad, BigDecimal>{
		private long totalPoints;
		private long pointsInCircle;
				
		@Override
		public void afterExecute(ActivityContext<WorkLoad, BigDecimal> ctx) throws Exception {
			BigDecimal total = new BigDecimal(totalPoints);
			BigDecimal inside = new BigDecimal(pointsInCircle);
			inside = inside.multiply(new BigDecimal(4));
			BigDecimal pi = inside.divide(total, MathContext.DECIMAL128);
			ctx.getOutput().add(pi);
			super.afterExecute(ctx);
		}

		@Override
		public int getOutputSize(ActivityContext<WorkLoad, BigDecimal> ctx) {
			return 0;
		}

		@Override
		public void input(ActivityContext<WorkLoad, BigDecimal> ctx, Lock<WorkLoad> inputLock) {
			WorkLoad wl = inputLock.get(0);
			totalPoints +=  wl.size;
			pointsInCircle += wl.result;
		}

		@Override
		public void output(ActivityContext<WorkLoad, BigDecimal> ctx, Lock<BigDecimal> outputLock) {}

	}

	public static BigDecimal calculate(ExecutionQueue fs, long randomPoints) throws Exception{
		final int workLoadCount = fs.workerCount();
		final long workLoadSize = randomPoints / workLoadCount;
		final long reminder = randomPoints % workLoadCount;
		WorkLoad[] load=new WorkLoad[workLoadCount];
		for(int i=0;i<workLoadCount;++i){
			load[i] = new WorkLoad(new Random(i), 
					i==0?workLoadSize+reminder:workLoadSize);
		}
		Channel<WorkLoad> input = new ArrayChannel<WorkLoad>(load);
		input.close();
		Pipeline<WorkLoad,BigDecimal> pipe= new Pipeline<WorkLoad,BigDecimal>(
				new Fork<WorkLoad, WorkLoad>(new PointCounter()), 
				new PiCalculator());
		Channel<BigDecimal> output = new ConcurrentArrayChannel<BigDecimal>(1);
		return fs.execute(pipe, input, output).waitForResult(5000000);
	}
	public static BigDecimal calculate(long randomPoints) throws Exception{
		long pointsInCircle=0;
		Random random = new Random(0);
		for(long i=0;i<randomPoints;++i){
			final double x = random.nextDouble();
			final double y = random.nextDouble();
			if (x*x + y*y < 1.0) ++ pointsInCircle;
		}
		BigDecimal total = new BigDecimal(randomPoints);
		BigDecimal inside = new BigDecimal(pointsInCircle);
		inside = inside.multiply(new BigDecimal(4));
		BigDecimal pi = inside.divide(total, MathContext.DECIMAL128);
		return pi;
	}
	
	private static long N = 10000000;
	
	@Test
	public void testSerial() throws Exception{
		calculate(N);
		long startTime = System.currentTimeMillis();
		System.out.println(calculate(N));
		long endTime = System.currentTimeMillis();
		long actualTime = endTime - startTime;
		System.out.println("Serial MonteCarlo time,ms = "+actualTime);
		
	}
	@Test
	public void testParallel() throws Exception {
		final CentralizedExecutionQueue queue = new CentralizedExecutionQueue();
		calculate(new CentralizedExecutionQueue(), N);
		long startTime = System.currentTimeMillis();
		System.out.println(calculate(new CentralizedExecutionQueue(), N));
		long endTime = System.currentTimeMillis();
		long actualTime = endTime - startTime;
		System.out.println("Cores="+queue.workerCount()+" Parallel Monte Carlo time,ms = "+actualTime);
	}
	
}
